package demo.base.connect;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;

public class OrderAddress extends Order {
    private String date;
    private String address;

    public OrderAddress(String orderId, String date, String address) {
        super(orderId);
        this.address = address;
        this.date = date;
    }

    @Override
    public String toString() {
        return JSON.toJSONString(this, SerializerFeature.WriteMapNullValue);
    }

    public static OrderAddress of(String orderId, String date, String address) {
        return new OrderAddress(orderId, date, address);
    }

    public OrderAddress coInit() {
        return new OrderAddress("", "", "");
    }

    public static OrderAddress empty() {
        return new OrderAddress("", "", "");
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public static OrderAddressMapFunction getMapFunction() {
        return new OrderAddressMapFunction();
    }

    private static class OrderAddressMapFunction extends RichMapFunction<String, OrderAddress> {

        @Override
        public OrderAddress map(String value) throws Exception {
            String[] splits = value.split(",");
            if (splits.length != 3) {
                System.out.println("OrderAddressMapFunction input error:" + value);
                return OrderAddress.empty();
            } else {
                return OrderAddress.of(splits[0], splits[1], splits[2]);
            }
        }
    }

    public static OrderAddressFlatMapFunction getFlatMapFunction() {
        return new OrderAddressFlatMapFunction();
    }

    private static class OrderAddressFlatMapFunction extends RichFlatMapFunction<String, String> {

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            String[] splits = value.split(",");
            if (splits.length != 3) {
                System.out.println("OrderAddressMapFunction input error:" + value);
            } else {
                JSONObject jsonObject = new JSONObject();
                String orderId = splits[0];
                String date = splits[1];
                String address = splits[2];

                jsonObject.put("orderId", orderId);
                jsonObject.put("date", date);
                out.collect(jsonObject.toJSONString());

                jsonObject.clear();

                jsonObject.put("orderId", orderId);
                jsonObject.put("address", address);
                //                jsonObject.put("address", null);
                out.collect(JSON.toJSONString(jsonObject, SerializerFeature.WriteMapNullValue));
            }
        }
    }
}
